package com.tongqin.distributed;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * @author yuantongqin
 * @date 2018/7/26 下午3:25
 */
public class KafkaProduceDemo  extends Thread{

    KafkaProducer<Integer,String> producer;
    private final String topic;

    /**
     * ProducerConfig.BOOTSTRAP_SERVERS_CONFIG :kafka集群上的地址
     */
    public KafkaProduceDemo(String topic) {
        Properties properties = new Properties();

        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"172.16.233.128:9092," +
                "172.16.233.129:9092,172.16.233.130:9092");

        properties.put(ProducerConfig.CLIENT_ID_CONFIG,"KafkaProduceDemo");
        properties.put(ProducerConfig.ACKS_CONFIG,"-1");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.IntegerSerializer");//这是序列化
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        this.producer = new KafkaProducer<Integer, String>(properties);
        this.topic = topic;
    }

    @Override
    public void run() {
        super.run();
        int num = 0;
        while (num <50){
            final String message = "have not have "+num;
            System.out.println(message);
//            producer.send(new ProducerRecord<Integer, String>(topic,message));
            producer.send(new ProducerRecord<Integer, String>(topic, message), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    System.out.println(metadata.offset()+"==offset==partition=="+metadata.partition());
                }
            });


            System.out.println("len");
            System.out.println();
            num ++;
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    public static void main(String[] args) {
//        new KafkaProduceDemo("test").start();
        int a = "哈哈哈".hashCode()%50;
        System.out.println(a);
    }
}
